package com.demo.reactor.mainsub;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

public class MainAcceptor implements Runnable{
    private ServerSocketChannel serverSocketChannel;
//    private AtomicInteger index = new AtomicInteger();
    private Integer index = 0;
    Selector[] selectors = new Selector[Runtime.getRuntime().availableProcessors()*2];
    SubReactor[] subReactors = new SubReactor[Runtime.getRuntime().availableProcessors()*2];
    Thread[] threads = new Thread[Runtime.getRuntime().availableProcessors()*2];



    public MainAcceptor(ServerSocketChannel serverSocketChannel) throws IOException {
        this.serverSocketChannel = serverSocketChannel;
        for (int i = 0; i < Runtime.getRuntime().availableProcessors()*2; i++) {
            selectors[i] = Selector.open();
            subReactors[i] = new SubReactor(selectors[i]);
            threads[i] = new Thread(subReactors[i]);
            threads[i].start();
        }

    }

    @Override
    public void run() {

        try {
            SocketChannel socketChannel = serverSocketChannel.accept();
            socketChannel.configureBlocking(false);
            selectors[index].wakeup();
            socketChannel.register( selectors[index], SelectionKey.OP_READ,new WorkerHandler(socketChannel));
            if(++index == threads.length){
                index = 0;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
